package com.shujia.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{Accumulator, SparkConf, SparkContext}

object Demo20Acc {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[8]").setAppName("map")
    val sc: SparkContext = new SparkContext(conf)

    val RDD1: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7))

    var sum: Int = 0
    RDD1.foreach(line => {

      /**
        * 在Executor中对一个普通变量做的任何修改都不会映射到Driver端
        *
        * 因为Executor端和Driver端属于不同的jvm进程
        *
        */

      sum += 1
      println(line)

    })

    println("sum:" + sum)

    /**
      * 累加器
      *
      *
      */

    //在Driver端定义一个累加器
    val acc: Accumulator[Int] = sc.accumulator(0)


    RDD1.foreach(i => {

      /**
        * 累加器
        * 1  在Executor端局部累加
        * 2   任务结束之后再Driver端合并结果
        *
        *
        * 注意
        * 1  累加器只能再Driver端定义
        * 2   累加器只能再Executor端累加
        * 3   累加器不能再Executor端读取
        * 4   累加器只能再Driver端读取
        *
        */

      //在Executor端累加
      acc.add(1)
      println(i)
    })

    //在Driver端读取累加结果
    println("acc=" + acc.value)


    /**
      * sparkd代码rdd不能嵌套使用
      *
      */
    /*RDD1.foreach(i => {

      RDD1.foreach(j => {
        println(i + j)
      })
    })*/


  }
}
